package cn.junechiu.learnrxjava;

import android.os.Bundle;
import android.support.annotation.Nullable;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.Button;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by junzhao on 2017/7/19.
 * 上游和下游分别是Observable和Observer,
 * 这次上游变成了Flowable, 下游变成了Subscriber, 但是水管之间的连接还是通过subscribe()
 * Flowable被压策略---BackpressureStrategy.ERROR
 * <p>
 * onSubscribe方法区别:
 * Disposable.dispose()
 * Subscription.cancel()
 * <p>
 * Flowable在设计的时候采用了一种新的思路也就是响应式拉取的方式来更好的解决上下游流速不均衡的问题
 * 所以我们把request当做是一种能力, 当成下游处理事件的能力, 下游能处理几个就告诉上游我要几个,
 * 这样只要上游根据下游的处理能力来决定发送多少事件,就不会造成一窝蜂的发出一堆事件来, 从而导致OOM
 * 只有当上游正确的实现了如何根据下游的处理能力来发送事件的时候, 才能达到这种效果, 如果上游根本不管下游的处理能力
 * <p>
 * 同步：
 * 下游没有调用request，上游不可能一直处于等待，所以会抛出一个异常，继续发送剩余的事件
 * 异步：
 * 上游却正确的发送了所有的事件呢? 这是因为在Flowable里默认有一个大小为128的水缸, 当上下游工作在不同的线程中时,
 * 上游就会先把事件发送到这个水缸中, 因此, 下游虽然没有调用request, 但是上游在水缸中保存着这些事件,
 * 只有当下游调用request时, 才从水缸里取出事件发给下游.
 * <p>
 * <p>
 * BackpressureStrategy.BUFFER 事件缓存没有大小限制
 * MissingBackpressureException异常
 */
public class FlowableActivity2 extends AppCompatActivity {

    private Button requestBtn;

    private Subscription subscription;

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_flowable2);
        requestBtn = (Button) findViewById(R.id.request_btn);
        requestBtn.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                subscription.request(1);  //拉取事件
            }
        });

//        sample1();
        sample2();
        sample3();
    }

    //正常发送，拉取
    public void sample1() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR)//选择被压异常
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(5); //拉取事件 不拉取会报异常
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d("FlowableActivity2", "integer: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.d("FlowableActivity2", "onError: " + t.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        Log.d("FlowableActivity2", "onComplete: ");
                    }
                });
    }

    public void sample2() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onNext(4);
                emitter.onNext(5);
                emitter.onNext(6);
            }
        }, BackpressureStrategy.ERROR)
                .subscribeOn(Schedulers.io())  //异步会缓存
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        subscription = s;
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d("FlowableActivity2", "integer: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.d("FlowableActivity2", "onError: " + t.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        Log.d("FlowableActivity2", "onComplete: ");
                    }
                });
    }

    public void sample3() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                for (int i = 0; i < 130; i++) {
                    e.onNext(i);
                    Log.d("FlowableActivity2", "FlowableEmitter: " + i);
                }
            }
        }, BackpressureStrategy.ERROR)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                    }

                    @Override
                    public void onNext(Integer integer) {
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.d("FlowableActivity2", "onError: " + t.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        Log.d("FlowableActivity2", "onComplete: ");
                    }
                });
    }
}
